//
// Created by benxb on 2021/11/16.
//

#include <chrono>
#include "mem/mem_data_table.h"
#include "common/logger.h"

namespace benxdb {

bool MemDataTable::AppendTuple(Tuple& tuple) {
    if (GetFreeRecordSlot() < 1) {
        return false;
    }
    log_manager_->AppendLog(tuple);
    // append record into its column store
    for (int i = 0; i < table_->GetColumnSize(); i++) {
        Value val;
        tuple.GetValue(table_->GetColumn(i), val);
        chunk_vec_[i]->AppendRecord(val);
    }
    record_count_++;
    return true;
}

bool MemDataTable::IsFull() const {
    return record_count_ == RECORD_COUNT_PER_CHUNK;
}

bool MemDataTable::AppendManyTuples(std::vector<Tuple>& data_vec) {
    // TODO
    return false;
}

void MemDataTable::Query(QueryPhrase query, std::vector<Tuple>& res) {
    std::vector<Selection> selections = query.GetSelections();
    std::vector<std::string> query_columns = query.GetQueryColumns();
    std::vector<int> column_offsets;
    for (const Selection& sel : selections) {
        int offset = table_->GetColumnIndex(sel.column_name);
        column_offsets.push_back(offset);
    }

    std::vector<int> query_column_offsets;
    for (const std::string& column : query_columns) {
        int offset = table_->GetColumnIndex(column);
        if (offset == -1) {
            continue;
        }
        query_column_offsets.push_back(offset);
    }

    for (int i = 0; i < record_count_; i++) {
        bool valid = true;
        for (int j = 0; j < column_offsets.size(); j++) {
            int offset = column_offsets[j];
            if (chunk_vec_[offset]->QueryIndex(i).Compare(selections[j].comp, selections[j].val) == CmpBool::CmpFalse) {
                valid = false;
            }
        }
        if (!valid) {
            continue;
        }
        std::vector<Value> query_values;
        for (int offset : query_column_offsets) {
            query_values.push_back(chunk_vec_[offset]->QueryIndex(i));
        }
        res.emplace_back(Tuple(query_values));
    }
}

std::shared_ptr<DataChunk> MemDataTable::GetChunkByIndex(int index) {
    if (index >= chunk_vec_.size()) {
        return nullptr;
    }
    return chunk_vec_[index];
}

MemDataTable::MemDataTable(const std::shared_ptr<Table> &table) : table_(table) {
    if (table == nullptr) {
        record_count_ = 0;
        return;
    }
    int column_size = table_->GetColumnSize();
    for (int i = 0; i < column_size; i++) {
        Column col = table->GetColumn(i);
        chunk_vec_.emplace_back(std::make_shared<DataChunk>(col.GetType()));
    }
    record_count_ = 0;
}

void MemDataTable::Clear() {
    LOG_DEBUG("Delete log file, id: %lu", GetLSN());
    log_manager_->DeleteLogFile();
    for (auto & chunk : chunk_vec_) {
        chunk->Clear();
    }
    record_count_ = 0;
}

MemDataTable::MemDataTable(const std::shared_ptr<Table> &table, uint64_t lsn) :
        MemDataTable(table)
{
    SetLSN(lsn);
}

bool MemDataTable::HasFreeRecordSlot() const {
    return record_count_ < RECORD_COUNT_PER_CHUNK;
}

int MemDataTable::GetFreeRecordSlot() const {
    return RECORD_COUNT_PER_CHUNK - record_count_;
}

void MemDataTable::SetLSN(lsn_t lsn) {
    lsn_ = lsn;
    log_manager_.reset(new LogManager(table_->GetName(), lsn));
    std::vector<Tuple> tuples;
    log_manager_->ReadAllLog(table_, tuples);
    AppendManyTuplesWithoutLog(tuples);
}

int MemDataTable::ColumnSize() {
    return chunk_vec_.size();
}

int MemDataTable::GetRecordCount() const {
    return record_count_;
}

void MemDataTable::AppendTupleWithoutLog(Tuple &tuple) {
    // append record into its column store
    for (int i = 0; i < table_->GetColumnSize(); i++) {
        Value val;
        tuple.GetValue(table_->GetColumn(i), val);
        chunk_vec_[i]->AppendRecord(val);
    }
    record_count_++;
}

bool MemDataTable::AppendManyTuplesWithoutLog(std::vector<Tuple> &data_vec) {
    for (auto& tuple : data_vec) {
        AppendTupleWithoutLog(tuple);
    }
}

MemDataTable::~MemDataTable() {
    timer_.StopAndPrint("MemDataTable Append records total.");
}

bool MemDataTable::AppendManyTuples(std::vector<Tuple> &tuples, int begin, int end) {
    timer_.Start();
    if (GetFreeRecordSlot() < end-begin+1) {
        return false;
    }
//    log_manager_->AppendManyLog(tuples, begin, end);
    for (int i = begin; i <= end; i++) {
        Tuple& tuple = tuples[i];
        for (int j = 0; j < table_->GetColumnSize(); j++) {
            Value val;
            tuple.GetValue(table_->GetColumn(j), val);
            chunk_vec_[j]->AppendRecord(val);
        }
    }
    record_count_ += end-begin+1;
    timer_.PauseAndPrint("MemTable Append records.");
    return true;
}

bool MemDataTable::AppendPointXXX(PointCloud<PointXYZI> &point_cloud, int begin_index, int end_index) {
    auto all_points = point_cloud.AllPoints();
    for (int i = begin_index; i <= end_index; i++) {
        auto& point = all_points->at(i);
        chunk_vec_[0]->AppendData(reinterpret_cast<const char*>(&point.x), 4);
        chunk_vec_[1]->AppendData(reinterpret_cast<const char*>(&point.y), 4);
        chunk_vec_[2]->AppendData(reinterpret_cast<const char*>(&point.z), 4);
        chunk_vec_[3]->AppendData(reinterpret_cast<const char*>(&point.intensity), 4);
    }
    return true;
}

void MemDataTable::QueryPointCloud(PointCloudQuery& query, PointCloud<PointXYZ>& res) {
    std::vector<std::string> query_columns = query.GetQueryAttributes();
    std::vector<int> column_offsets;
    if (chunk_vec_.size() < 3) {
        return;
    }

    auto x_chunk = chunk_vec_[0];
    auto y_chunk = chunk_vec_[1];
    auto z_chunk = chunk_vec_[2];

    for (uint i = 0; i < record_count_; i++) {
        auto x = x_chunk->QueryIndex(i).GetAs<float>();
        auto y = y_chunk->QueryIndex(i).GetAs<float>();
        if (!query.Satisfy(x, y)) {
            continue;
        }
        auto z = z_chunk->QueryIndex(i).GetAs<float>();
        res.Append({i, x, y, z});
    }
}

bool MemDataTable::AppendPointData(const char *point_data, int size) {
    if (!GetFreeRecordSlot()) {
        return false;
    }

    log_manager_->AppendPointLog(point_data, size);
    int offset = 0;
    for (int i = 0; i < table_->GetColumnSize(); i++) {
        int len = table_->GetColumn(i).GetLength();
        chunk_vec_[i]->AppendData(point_data + offset, len);
    }
    record_count_ += 1;
    return true;
}

bool MemDataTable::IsEmpty() const {
    return record_count_ == 0;
}

bool MemDataTable::AppendPointXXX(PointCloud<PointSTXYZC> &point_cloud, int begin_index, int end_index) {
    auto all_points = point_cloud.AllPoints();
    for (int i = begin_index; i <= end_index; i++) {
        auto& point = all_points->at(i);
        chunk_vec_[0]->AppendData(reinterpret_cast<const char *>(&point.sweep), sizeof(point.sweep));
        chunk_vec_[1]->AppendData(reinterpret_cast<const char *>(&point.timestamp), sizeof(point.timestamp));
        chunk_vec_[2]->AppendData(reinterpret_cast<const char*>(&point.x), 4);
        chunk_vec_[3]->AppendData(reinterpret_cast<const char*>(&point.y), 4);
        chunk_vec_[4]->AppendData(reinterpret_cast<const char*>(&point.z), 4);
        chunk_vec_[5]->AppendData(reinterpret_cast<const char*>(&point.classification), sizeof(point.classification));
    }
}

bool MemDataTable::AppendPointXXX(PointCloud<PointSTXYZ> &point_cloud, int begin_index, int end_index) {
    auto all_points = point_cloud.AllPoints();
    for (int i = begin_index; i <= end_index; i++) {
        auto& point = all_points->at(i);
        chunk_vec_[0]->AppendData(reinterpret_cast<const char *>(&point.sweep), sizeof(point.sweep));
        chunk_vec_[1]->AppendData(reinterpret_cast<const char *>(&point.timestamp), sizeof(point.timestamp));
        chunk_vec_[2]->AppendData(reinterpret_cast<const char*>(&point.x), 4);
        chunk_vec_[3]->AppendData(reinterpret_cast<const char*>(&point.y), 4);
        chunk_vec_[4]->AppendData(reinterpret_cast<const char*>(&point.z), 4);
    }
}

}